package org.witsoft.localmq.impl;

import org.witsoft.localmq.FileReader;
import org.witsoft.localmq.FileWriter;
import org.witsoft.localmq.MessageManager;
import org.witsoft.localmq.OffsetManager;
import org.witsoft.localmq.model.Offset;
import org.witsoft.localmq.model.Message;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.Synchronized;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @ClassName: MqManagerImpl
 * @Description:
 * @Author: Colin
 * @Date: 2020/12/26 2:24 PM
 * @Version 1.0
 **/
@RequiredArgsConstructor
public class MessageManagerImpl implements MessageManager {

    @NonNull
    private OffsetManager offsetManager;

    @NonNull
    private FileWriter fileWriter;

    @NonNull
    private FileReader fileReader;

    @NonNull
    private QueuedMessageBroker messageBroker;

    @Override
    public void produce(byte[]... batchData) throws IOException{
        List<Message> messageList = new ArrayList<>(batchData.length);
        for (byte[] data : batchData) {
            // 产生偏移量
            int dataSize = data.length;
            Offset offset = offsetManager.nextProduceOffset(dataSize);
            // 写入数据
            Message message = Message.builder()
                    .offset(offset.getCurrentOffset())
                    .segment(offset.getSegmentId())
                    .position(offset.getPosition())
                    .data(data)
                    .build();
            messageList.add(message);
        }
        if(messageBroker != null){
            messageBroker.produce(messageList);
        }else {
            fileWriter.write(messageList);
        }
    }

    @Override
    @Synchronized
    public List<Message> consume(int size) throws IOException{
        List<Offset> offsetList = new ArrayList<>(size);
        long startOffset = offsetManager.getConsumeOffset();
        for (int i = 0; i < size ; i++) {
            Offset curOffset = offsetManager.getOffset(startOffset + i);
            if(curOffset == null){
                break;
            }
            offsetList.add(curOffset);
        }
        List<Message> messageList = fileReader.read(offsetList);
        if(!offsetList.isEmpty() && !messageList.isEmpty()) {
            Offset lastOffset = offsetList.get(offsetList.size() - 1);
            long currentOffset = lastOffset.getCurrentOffset();
            offsetManager.commitConsumeOffset(currentOffset  + 1);
        }
        return messageList;
    }

    @Override
    public List<Message> poll(int offset, int size) throws IOException {
        List<Offset> offsetList = new ArrayList<>(size);
        // 安
        long startOffset = offset >= 0 ? offset : offsetManager.getConsumeOffset();
        for (int i = 0; i < size ; i++) {
            Offset curOffset = offsetManager.getOffset(startOffset + i);
            if(curOffset == null){
                break;
            }
            offsetList.add(curOffset);
        }
        return fileReader.read(offsetList);
    }

    @Override
    public void commitConsumeOffset(int offset) throws IOException {
        offsetManager.commitConsumeOffset(offset  + 1);
    }

    public void purge() throws IOException{
        offsetManager.purge();
    }
}
